【小ネタ】Lambda関数に画像データを渡す
(2021/6/6 内容を修正しました)
カフェチームの山本です。
現在、カフェのシステムでは、機械学習を用いて、カメラを用いて動画を撮影し、商品の前にいる人物の骨格や手を検出することで、どのユーザがどの商品を取り出したかを判定しています。
そのために、エッジデバイスからクラウドに動画を送信し、クラウド側で様々な処理を実行する、という構成を検討しており、最近の記事で、Kinesis Video Streamsから動画を取り出す方法、Lambdaレイヤを使ってTensorFlow Liteをインポートし、Lambda関数で推論処理を実行する方法ついて記載しました。
【Kinesis Video Streams】Lambda関数でKinesis Video Streamsから動画を取得する | DevelopersIO
Lambda関数でTensorFlow Liteをインポートし、推論処理を実行する | DevelopersIO
今回は、取り出した動画から画像を切り出して、推論処理を実行するLambda関数に画像データを渡す方法として、入力となるpayloadに直接入れ込む方法について記載します。他の方法としても、S3バケットに保存してキーを渡す・トリガーで実行する、SQSキューにメッセージを送信する、といった方法があります。なので、今回の方法は、他のリソースを用意するのが面倒だったり、他のサービスを介する時間が気になる場合に、一つの選択肢として考えてみてください。
実装コード
今回実装したコードは以下のとおりです。
入力方法
frameには画像データ(np.ndarray)を、lambda_function_nameには実行したいLambda関数の名前を入力する想定です(実行する際には、lambda_function_nameをinvokeする権限があるか、確認してください)。
import cv2 import base64 import json import boto3 def invoke_infer_lambda_frame(frame, lambda_function_name): # compress image filepath_img = f"/tmp/img.png" cv2.imwrite(filepath_img, frame) img_bytes = open(filepath_img, "rb").read() # convert to b64-str img_b64 = base64.b64encode(img_bytes).decode("utf8") # invoke lambda data = { "img_b64": img_b64, "filename": "img.png", } payload = json.dumps(data) ret = boto3.client("lambda").invoke( FunctionName=lambda_function_name, InvocationType='RequestResponse', # Event(非同期) or RequestResponse(同期) Payload=payload, )
動作としては、以下の通りです
- 画像をファイルとして出力し、ファイルをバイナリとして読み込みなおす。これによってデータサイズを小さく(圧縮)します。(他にも良い方法があるかもしれません)
- バイナリデータをbase64で変換した後、文字列に変換する。これによって、そのままpayloadに加えることはできないバイナリデータを、送信できる文字列に変換します。このとき、データサイズは、前のステップで圧縮した後の画像のサイズより、4/3倍のサイズに大きくなります。
- payload用にjson形式で変換し、Lambda関数をinvokeする。invokeする際、画像のデータサイズによって、選べるInvocationTypeが異なります。Lambdaの制限として、payloadが256KB以下ならEvent(非同期)・RequestResponse(同期)の両方を選ぶことができ、256KBを超える場合はRequestResponse(同期)しか選べません(6MBを超える場合は、どちらも実行できません)。先程のbase64の変換で大きくなった後のサイズで決まるので注意が必要です。
受け取る方法
処理は以下のとおりです。先程の処理を逆順にたどる形です。
- eventからパラメータを取り出す
- 文字列からバイナリに戻し、base64で画像データにもどす。
- ファイルとして出力して読み直し、もとのframeに戻す。(今回は受け取ったfilenameをそのまま利用していますが、その必要はありませんが、圧縮時に利用したときと、同じ拡張子で出力する必要があります。)
import cv2 import base64 def lambda_handler(event, context): # get parameters img_b64 = event["img_b64"] filename = event["filename"] # convert to bytes img_bytes = base64.b64decode(img_b64.encode("utf8")) # decompress image with open("/tmp/" + filename, "wb") as file: file.write(img_bytes) img = cv2.imread(filepath_img) # ...
まとめ
Lambdaに画像データを渡す方法として、base64で文字列にデコードする方法を利用しました。payloadのサイズによって、利用できるInvocationTypeが異なるので、注意が必要です。